package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constans.ScheduleConstants;
import com.heima.common.redis.RedisCacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;


@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Resource
    private RedisCacheService redisCacheService;

    @Resource
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 添加任务
     *
     * @param task 任务对象
     * @return 任务id
     */
    @Override
    public void addTask(Task task) {
        threadPoolTaskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                //1.新增任务到数据库表中，taskinfo和taskinfologs表
                boolean flag = saveTaskToDB(task);
                if(!flag){
                    return;
                }
                //2.添加任务数据到redis中
                saveTaskToRedis(task);
            }
        });
    }


    /**
     * 保存数据到redis中
     * @param task
     */
    private void saveTaskToRedis(Task task) {
        //定义一个key前缀
        String keysuffix=task.getTaskType()+"_"+task.getPriority();

        //2.1 当执行时间小于等于当前系统时间的时候，存入到list中
        if(task.getExecuteTime()<=System.currentTimeMillis()){
            redisCacheService.lLeftPush(ScheduleConstants.TOPIC+keysuffix, JSON.toJSONString(task));
        }
        //2.2 当执行时间小于等于当前系统时间+5分钟，存入到zset中
        //获取当前系统时间+5分钟的毫秒数
        Calendar calendar=Calendar.getInstance();//jdk8提供的一个日历类
        calendar.add(Calendar.MINUTE,5);//当前系统时间+5分钟

        if( task.getExecuteTime()>System.currentTimeMillis() &&task.getExecuteTime()<=calendar.getTimeInMillis()){
            redisCacheService.zAdd(ScheduleConstants.FUTURE+keysuffix, JSON.toJSONString(task),task.getExecuteTime());
        }
    }

    /**
     * .新增任务到数据库表中，taskinfo和taskinfologs表
     * @param task
     */
    private boolean saveTaskToDB(Task task) {
        //1.新增数据到taskinfo表中
        Taskinfo taskinfo=new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoMapper.insert(taskinfo);

        //为了保证后面可以使用task_id
        task.setTaskId(taskinfo.getTaskId());

        //2.新增数据到taskinfologs表中
        TaskinfoLogs taskinfologs=new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfologs);
        taskinfologs.setVersion(1);
        taskinfologs.setStatus(ScheduleConstants.SCHEDULED);
        int i = taskinfoLogsMapper.insert(taskinfologs);
        return i==1;
    }


    /**
     * 消费任务
     *
     * @param type     类型
     * @param priority 优先级
     */
    @Override
    public Task pullTask(int type, int priority) {
        Future<Task> future = threadPoolTaskExecutor.submit(new Callable<Task>() {

            @Override
            public Task call() throws Exception {
                //定义一个key前缀
                String keysuffix = type + "_" + priority;
                //1.从list中取出数据并删除
                String taskStr = redisCacheService.lRightPop(ScheduleConstants.TOPIC + keysuffix);
                if (taskStr != null) {
                    //字符串转对象
                    Task task = JSON.parseObject(taskStr, Task.class);
                    //2.删除taskinfo表数据
                    taskinfoMapper.deleteById(task.getTaskId());

                    //3.修改taskinfologs日志表数据
                    TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(task.getTaskId());
                    if (taskinfoLogs != null) {
                        //1.改状态
                        taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
                        //2.改乐观锁,按道理来说我们应该version+1,但是mp集成了乐观锁
                        //taskinfoLogs.setVersion();
                        taskinfoLogsMapper.updateById(taskinfoLogs);
                    }

                    return task;
                }
                return null;
            }
        });
        try {
            return future.get();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    /**
     * 定时刷新数据从zset中同步到list中
     * 注意：当前这个定时任务方法，不会被任何接口给调用，自己完成定时任务，所以此方法没有参数
     * 第二，由于是定时任务方法同样也不会有任何返回值
     */
    //定时器，每分钟执行一次
   // @Scheduled(cron = "0 0/1 * * * ?")//执行定时任务的注解，cron表达式表示具体什么时候执行定时任务
    @Override
    @Scheduled(cron = "*/1 * * * * ?")//为了测试每秒钟执行一次
    public void refresh() {
        threadPoolTaskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                String token = redisCacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
                if(StringUtils.isNotBlank(token)){//表示加锁成功
                    System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");

                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //1.获取zset中的所有的key,然后判断当前数据是否到期
                    Set<String> zsetkeys = redisCacheService.scan("future_*");
                    for (String zsetkey : zsetkeys) {//zsetkey=future_100_50
                        //根据score值(执行时间)小于等于当前系统时间，则取出来,taskZSet表示到期的数据
                        Set<String> taskZSet = redisCacheService.zRangeByScore(zsetkey, 0, System.currentTimeMillis());
                        //2.如果到期了，则提取出数据，并删除数据，然后同步到list队列中
                        if (taskZSet!=null && taskZSet.size()>0){
                            /**
                             *  同步数据
                             *  refreshWithPipeline(String future_key,String topic_key,Collection<String> values)
                             *  参数1表示未来数据的key
                             *  参数2表示list队列的key
                             *  参数3表示任务信息集合
                             */
                            String future_key=zsetkey;
                            String topic_key=ScheduleConstants.TOPIC+zsetkey.split(ScheduleConstants.FUTURE)[1];
                            redisCacheService.refreshWithPipeline(future_key,topic_key,taskZSet);
                            System.out.println("成功的将" + future_key + "下的当前需要执行的任务数据刷新到" + topic_key + "下");
                        }
                    }

                }
            }
        });

    }

    /**
     * 定时同步数据从数据库中到redis中
     */
    @Scheduled(cron = "0 0/5 * * * ?")//每5分钟执行一次
    @Override
    public void reloadData() {
        log.info("数据库数据同步到缓存");
        //1.清除缓存中的数据，是有争议，
       /* Set<String> listKeys = redisCacheService.scan("topic_*");
        Set<String> zsetKeys = redisCacheService.scan("future_*");
        redisCacheService.delete(listKeys);
        redisCacheService.delete(zsetKeys);*/

        //2.查询数据库表中的taskinfo，执行时间小于等于当前时间+5分钟的数据
        Calendar calendar=Calendar.getInstance();//jdk8提供的一个日历类
        calendar.add(Calendar.MINUTE,5);//当前系统时间+5分钟

        LambdaQueryWrapper<Taskinfo> queryWrapper= Wrappers.lambdaQuery();
        queryWrapper.le(Taskinfo::getExecuteTime,calendar.getTimeInMillis());
        List<Taskinfo> taskinfos = taskinfoMapper.selectList(queryWrapper);

        //3.保存数据到redis中，有可能到list，也有可能到zset中
        for (Taskinfo taskinfo : taskinfos) {
            Task task=new Task();
            BeanUtils.copyProperties(taskinfo,task);
            task.setExecuteTime(taskinfo.getExecuteTime().getTime());
            this.saveTaskToRedis(task);
        }
    }
}
